package com.gis.aoitest.window;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

public class TumblingEvnetWindows3 {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 输入流
        DataStreamSource<String> socketStream = env.socketTextStream("localhost", 9999);

        // 处理数据
        socketStream.map(line -> new Tuple3<String, Long, Double>(line.split(",")[0], Long.parseLong(line.split(",")[1]), Double.parseDouble(line.split(",")[2])))
                .returns(Types.TUPLE(Types.STRING, Types.LONG, Types.DOUBLE))
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Double>>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(Tuple3<String, Long, Double> element) {
//                        System.out.println(Thread.currentThread() + element.f0 + element.f1);
                        return element.f1 * 1000L;
                    }
                })
                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .minBy(2)
                .reduce(new ReduceFunction<Tuple3<String, Long, Double>>() {
                    @Override
                    public Tuple3<String, Long, Double> reduce(Tuple3<String, Long, Double> t2, Tuple3<String, Long, Double> t1) throws Exception {
                        System.out.println(t2 + "::" + t1);
                        System.out.println( Thread.currentThread() + "::" + Thread.currentThread().getName() + "::" + t1.f1);
                        return new Tuple3<>(t1.f0 + Math.min(t1.f1, t2.f1), Math.max(t1.f1, t2.f1), Math.min(t1.f2, t2.f2));
                    }
                })
                .print("stream");

        env.execute();
    }
}
